package com.stp.queue.delay;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

@Slf4j
public class DelayQueueUtil {
    private static final Map<String, Consumer<?>> CONSUMER_MAP = new ConcurrentHashMap<>();

    private static final AtomicBoolean STARTING = new AtomicBoolean();

    /**
     * 延迟队列
     */
    private static final DelayQueue<DelayQueueMessage<?>> DELAY_QUEUE = new DelayQueue<>();

    private static final int CORE_POOL_SIZE = 4;
    private static final int MAXIMUM_POOL_SIZE = 10;
    private static final long KEEP_ALIVE_TIME = 20;
    private static final TimeUnit UNIT = TimeUnit.SECONDS;
    private static final int MAXIMUM_ARRAY_SIZE = 10;
    private static final ThreadFactory NAMED_FACTORY = new ThreadFactoryBuilder().setNamePrefix("java_delay_thread_%d").build();

    /**
     * 执行读取任务的线程池
     */
    private static final ExecutorService THREAD_POOL = new ThreadPoolExecutor(
            CORE_POOL_SIZE,
            MAXIMUM_POOL_SIZE,
            KEEP_ALIVE_TIME,
            UNIT,
            new ArrayBlockingQueue<>(MAXIMUM_ARRAY_SIZE),
            NAMED_FACTORY);


    /**
     * 提交一个延迟消息
     * @param uuid 消息的uuid
     * @param msg 消息对象
     * @param consumer 延迟到期后回到方法
     * @param delayTime 延迟时间，毫秒
     * @param <T> 消息对象类型
     * @return true: 提交成功
     */
    public static <T> boolean submit(String uuid, T msg, Consumer<T> consumer, long delayTime) {
        DelayQueueMessage<T> DelayQueueMessage = new DelayQueueMessage<>(uuid, msg, delayTime);
        addTask(uuid, consumer);

        return DELAY_QUEUE.offer(DelayQueueMessage);
    }

    /**
     * 取消一个延迟消息
     * @param uuid 消息的uuid
     * @return true: 取消成功
     */
    public static boolean cancel(String uuid) {
        return CONSUMER_MAP.remove(uuid) != null;
    }

    /**
     * 添加任务，懒加载开启消费线程
     * @param uuid 消息的uuid
     * @param consumer 回调方法
     * @param <T> 消息对象类型
     */
    private static <T> void addTask(String uuid, Consumer<T> consumer) {
        CONSUMER_MAP.put(uuid, consumer);

        // STARTING 是false，则开启监听队列的线程
        if (!STARTING.compareAndSet(false, true)) {
            return;
        }
        THREAD_POOL.execute(() -> {
            while (STARTING.get()) {
                try {
                    DelayQueueMessage<T> DelayQueueMessage = (DelayQueueMessage<T>) DELAY_QUEUE.take();
                    // 只有当map里面有该uuid对应的消息，才执行回调方法
                    if (CONSUMER_MAP.containsKey(DelayQueueMessage.getUuid())) {
                        // 执行回调方法
                        execCall(consumer, DelayQueueMessage);
                    }
                } catch (InterruptedException e) {
                    STARTING.set(false);
                }
            }
        });
    }

    private static <T> void execCall(Consumer<T> consumer, DelayQueueMessage<T> DelayQueueMessage) {
        CONSUMER_MAP.remove(DelayQueueMessage.getUuid());
        THREAD_POOL.execute(() -> consumer.accept(DelayQueueMessage.getBody()));
    }
}
